

#include "SerialMemoryManager.h"


SerialMemoryManager::SerialMemoryManager(MemoryTable *parMemoryTable, int parCollectorThreshold, int parNumClusters, float parLastThresold){

	//timer = new Timer();

	memoryTable = parMemoryTable;
	memoryTableToReset = false;
	collector = new Collector();
	encoder = new Encoder(parNumClusters);
	collectorThreshold = parCollectorThreshold;
	lastThreshold = parLastThresold;
	subscriptionMapMutex = new pthread_mutex_t;
	pthread_mutex_init(subscriptionMapMutex, NULL);
	eventMapMutex = new pthread_mutex_t;
	pthread_mutex_init(eventMapMutex, NULL);
	memoryTableMutex = new pthread_mutex_t;
	pthread_mutex_init(memoryTableMutex, NULL);
	memoryTableToResetMutex = new pthread_mutex_t;
	pthread_mutex_init(memoryTableToResetMutex, NULL);
//	updatingThread = new pthread_t;
//	pthread_create(updatingThread, NULL, serial_manager::startUpdatingThread, (void *) this);

	updatingCounter = 0;
	updatingThreshold = 40;

}

SerialMemoryManager::~SerialMemoryManager(){

	delete collector;
	delete encoder;
	pthread_mutex_destroy(subscriptionMapMutex);
	pthread_mutex_destroy(eventMapMutex);
	pthread_mutex_destroy(memoryTableMutex);
	pthread_mutex_destroy(memoryTableToResetMutex);
	delete subscriptionMapMutex;
	delete eventMapMutex;
	delete memoryTableMutex;
	delete memoryTableToResetMutex;
//	pthread_cancel(*updatingThread);
//	delete updatingThread;

}


void SerialMemoryManager::addEvent(PubPkt *pkt){

//	updatingCounter++;

	string complexType = encoder->encode(pkt);
	cout << "Memory manager: Serial: message clustered in " << complexType  << "." << endl;

	pthread_mutex_lock(eventMapMutex);
	pthread_mutex_lock(memoryTableMutex);
	boost::unordered_map<string, int>::iterator it = eventMap.find(complexType);
	if (it==eventMap.end()) {
		eventMap.insert(make_pair(complexType, 1));
	} else {
		it->second++;
	}
	cout << "Memory manager: Serial: start storing message..." << endl;
	memoryTable->computeInsertion(pkt, complexType);
	cout << "Memory manager: Serial: message stored." << endl;
	pthread_mutex_unlock(memoryTableMutex);
	pthread_mutex_unlock(eventMapMutex);

}

void SerialMemoryManager::addSubscription(BackSubPkt *pkt){

	updatingCounter++;

	collector->inspectSubscription(pkt);
	set<string> clustersMatched = encoder->encode(pkt);

	pthread_mutex_lock(subscriptionMapMutex);
	cout << "Memory manager: Serial: subscription clustered in ";
	for (set<string>::iterator it=clustersMatched.begin(); it!=clustersMatched.end(); it++) {
		string complexType = *it;
		cout << complexType << " ";
		boost::unordered_map<string, SubscribtionInfo>::iterator it2 = subscriptionMap.find(complexType);
		if (it2==subscriptionMap.end()) {
			SubscribtionInfo subInfo;
			subInfo.windowsList.push_back(pkt->getWindow());
			if (pkt->getOccurrence() == LAST){
				subInfo.countLast = 1;
			} else {
				subInfo.countLast = 0;
			}
			subscriptionMap.insert(make_pair(complexType, subInfo));
		} else {
			it2->second.windowsList.push_back(pkt->getWindow());
			if (pkt->getOccurrence() == LAST){
				it2->second.countLast++;
			}
		}
	}
	cout << "." << endl;


	if (updatingCounter == updatingThreshold){
		pthread_mutex_lock(eventMapMutex);
//		pthread_mutex_lock(subscriptionMapMutex);
		pthread_mutex_lock(memoryTableMutex);
		pthread_mutex_lock(memoryTableToResetMutex);
		cout << "----------------------------------------------------------------" << endl;
		cout << "Memory manager: Serial: start memory table updating..." << endl;
		if (memoryTableToReset){
			memoryTable->resetTable();
			memoryTableToReset = false;
		}
		pthread_mutex_unlock(memoryTableToResetMutex);
		for(boost::unordered_map<string, SubscribtionInfo>::iterator it = subscriptionMap.begin(); it!=subscriptionMap.end(); it++){
			string complexType = it->first;
			cout << "cluster: " << complexType << "\t";

			int eventsCount;
			boost::unordered_map<string, int>::iterator itEvent = eventMap.find(complexType);
			if (itEvent != eventMap.end()){
				eventsCount = itEvent->second;
				itEvent->second = 0;
			} else {
				eventsCount = 0;
			}
			cout << "events: " << eventsCount << "\t";
			vector<float> windowsVector = it->second.windowsList;
			int subsCount = windowsVector.size();
			cout << "subscriptions: " << subsCount << "\t";
			sort(windowsVector.begin(), windowsVector.end());
			float median = windowsVector[(int)windowsVector.size()/2];
			cout << "window: " << median << "\t";
			bool resettable;
			float percentage = it->second.countLast / subsCount;
			if (percentage < lastThreshold){
				resettable = true;
			} else {
				resettable = false;
			}
			cout << "resettable: " << util::boolToString(resettable) << "\t" << endl;
			memoryTable->updateTable(complexType, subsCount, median, resettable, eventsCount);
		}
		// *** SE POSSIBILE DA OTTIMIZZARE LA SEGUENTE *** //
		string complexType = "unknown";
		int eventsCount = 0;
		cout << "cluster: " << complexType << "\t\t";
		for(boost::unordered_map<string, int>::iterator it = eventMap.begin(); it!=eventMap.end(); it++){
			boost::unordered_map<string, SubscribtionInfo>::iterator itSub = subscriptionMap.find(it->first);
			if (itSub == subscriptionMap.end()){
				eventsCount += it->second;
				it->second = 0;
			}
		}
		cout << "events: " << eventsCount << "\t";
		int subsCount = 1;
		cout << "subscriptions: " << subsCount << "\t";
		float median = 1000;								// *** VALORE DA DECIDERE!! *** //
		cout << "window: " << median << "\t";
		bool resettable = true;
		cout << "resettable: " << util::boolToString(resettable) << "\t" << endl;
		memoryTable->updateTable(complexType, subsCount, median, resettable, eventsCount);
		// *** *** //
		memoryTable->fixTable();
		//timer->stop();
		cout << "Memory manager: Serial: memory table updated " << endl; //in " << timer->getElapsedTimeInMilliSec() << " ms." << endl;
		cout << "----------------------------------------------------------------" << endl << endl;

		updatingCounter = 0;

		pthread_mutex_unlock(memoryTableMutex);
//		pthread_mutex_unlock(subscriptionMapMutex);
		pthread_mutex_unlock(eventMapMutex);
	}


	if (collector->getCount() == collectorThreshold){
		cout << endl;
		cout << "++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++" << endl;
		cout << "Memory manager: Serial: start encoder updating..." << endl;
		pthread_mutex_lock(eventMapMutex);
	//	pthread_mutex_lock(subscriptionMapMutex);
		//timer->start();
		encoder->clear();
		boost::unordered_map<int, boost::unordered_map<string, vector<Type> > > collectorMap = collector->getMap();
		for (boost::unordered_map<int, boost::unordered_map<string, vector<Type> > >::iterator it=collectorMap.begin(); it!=collectorMap.end(); it++) {
			encoder->addEntry(it->first, it->second);
		}
//		cout << "Entry tutte inserite!" << endl;
		subscriptionMap.clear();
		eventMap.clear();
		collector->reset();
		pthread_mutex_lock(memoryTableToResetMutex);
		memoryTableToReset = true;
		//timer->stop();
		pthread_mutex_unlock(memoryTableToResetMutex);
//		pthread_mutex_unlock(subscriptionMapMutex);
		pthread_mutex_unlock(eventMapMutex);
		cout << "Memory manager: Serial: encoder updated" << endl; // in " << timer->getElapsedTimeInMilliSec() << " ms." << endl;
		cout << "++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++" << endl << endl;
	}

	pthread_mutex_unlock(subscriptionMapMutex);
}


void SerialMemoryManager::queryEventSet(BackSubPkt *pkt, Outbox &outbox){
	pthread_mutex_lock(memoryTableMutex);
	cout << "Memory manager: Serial: starting messages extraction..." << endl;
	memoryTable->computeExtraction(pkt, outbox.pkts);
	cout << "Memory manager: Serial: extraction terminated." << endl;
	pthread_mutex_unlock(memoryTableMutex);
}


void SerialMemoryManager::updateTable(){


//	sleep(5);
	usleep(50*1000);
	while (true) {
		//	cout << endl;
		//timer->start();
		pthread_mutex_lock(eventMapMutex);
		pthread_mutex_lock(subscriptionMapMutex);
		pthread_mutex_lock(memoryTableMutex);
		pthread_mutex_lock(memoryTableToResetMutex);
		cout << "----------------------------------------------------------------" << endl;
		cout << "Memory manager: Serial: start memory table updating..." << endl;
		if (memoryTableToReset){
			memoryTable->resetTable();
			memoryTableToReset = false;
		}
		pthread_mutex_unlock(memoryTableToResetMutex);
		for(boost::unordered_map<string, SubscribtionInfo>::iterator it = subscriptionMap.begin(); it!=subscriptionMap.end(); it++){
			string complexType = it->first;
			cout << "cluster: " << complexType << "\t";

			int eventsCount;
			boost::unordered_map<string, int>::iterator itEvent = eventMap.find(complexType);
			if (itEvent != eventMap.end()){
				eventsCount = itEvent->second;
				//itEvent->second = 0;
			} else {
				eventsCount = 0;
			}
			cout << "events: " << eventsCount << "\t";
			vector<float> windowsVector = it->second.windowsList;
			int subsCount = windowsVector.size();
			cout << "subscriptions: " << subsCount << "\t";
			sort(windowsVector.begin(), windowsVector.end());
			float median = windowsVector[(int)windowsVector.size()/2];
			cout << "window: " << median << "\t";
			bool resettable;
			float percentage = it->second.countLast / subsCount;
			if (percentage < lastThreshold){
				resettable = true;
			} else {
				resettable = false;
			}
			cout << "resettable: " << util::boolToString(resettable) << "\t" << endl;
			memoryTable->updateTable(complexType, subsCount, median, resettable, eventsCount);
		}
		//*** SE POSSIBILE DA OTTIMIZZARE LA SEGUENTE ***//
		string complexType = "unknown";
		int eventsCount = 0;
		cout << "cluster: " << complexType << "\t\t";
		for(boost::unordered_map<string, int>::iterator it = eventMap.begin(); it!=eventMap.end(); it++){
			boost::unordered_map<string, SubscribtionInfo>::iterator itSub = subscriptionMap.find(it->first);
			if (itSub == subscriptionMap.end()){
				eventsCount += it->second;
				//it->second = 0;
			}
		}
		cout << "events: " << eventsCount << "\t";
		int subsCount = 1;
		cout << "subscriptions: " << subsCount << "\t";
		float median = 30000;								//*** VALORE DA DECIDERE!! ***/
		cout << "window: " << median << "\t";
		bool resettable = true;
		cout << "resettable: " << util::boolToString(resettable) << "\t" << endl;
		memoryTable->updateTable(complexType, subsCount, median, resettable, eventsCount);
		//*** ***//
		memoryTable->fixTable();
		//timer->stop();
		cout << "Memory manager: Serial: memory table updated " << endl; //in " << timer->getElapsedTimeInMilliSec() << " ms." << endl;
		cout << "----------------------------------------------------------------" << endl << endl;
		pthread_mutex_unlock(memoryTableMutex);
		pthread_mutex_unlock(subscriptionMapMutex);
		pthread_mutex_unlock(eventMapMutex);
		usleep(50*1000);
	}
}


void * serial_manager::startUpdatingThread(void *input){
	SerialMemoryManager *manager = (SerialMemoryManager*) input;
	manager->updateTable();
	pthread_exit(NULL);
	return 0;
}










